package com.xxl.job.consume.kafka;

import com.xxl.job.consume.common.KafkaTopicConstant;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;

import java.util.List;
import java.util.Optional;

public class KafkaConsumer {

    @Autowired
    private KafkaConsumeService kafkaConsumeService;

    /**
     * 环控
     * @param records
     */
    @KafkaListener(id="hkCurrent0",topics = KafkaTopicConstant.HK_ALARM_TOPIC_MESSAGE,containerFactory = "hkListenerContainerFactory")
    public void hkListenPartition0(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                kafkaConsumeService.dealHkMessage(record);
            }
        }
    }
    /**
     * 环控
     * @param records
     */
    @KafkaListener(id="hkCurrent1",topics = KafkaTopicConstant.HK_ALARM_TOPIC_MESSAGE,containerFactory = "hkListenerContainerFactory")
    public void hkListenPartition1(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                kafkaConsumeService.dealHkMessage(record);
            }
        }
    }
    /**
     * 环控
     * @param records
     */
    @KafkaListener(id="hkCurrent2",topics = KafkaTopicConstant.HK_ALARM_TOPIC_MESSAGE,containerFactory = "hkListenerContainerFactory")
    public void hkListenPartition2(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                kafkaConsumeService.dealHkMessage(record);
            }
        }
    }
    /**
     * 环控
     * @param records
     */
    @KafkaListener(id="hkCurrent3",topics = KafkaTopicConstant.HK_ALARM_TOPIC_MESSAGE,containerFactory = "hkListenerContainerFactory")
    public void hkListenPartition3(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                kafkaConsumeService.dealHkMessage(record);
            }
        }
    }
    /**
     * 环控
     * @param records
     */
    @KafkaListener(id="hkCurrent4",topics = KafkaTopicConstant.HK_ALARM_TOPIC_MESSAGE,containerFactory = "hkListenerContainerFactory")
    public void hkListenPartition4(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                kafkaConsumeService.dealHkMessage(record);
            }
        }
    }
    /**
     * 环控
     * @param records
     */
    @KafkaListener(id="hkCurrent5",topics = KafkaTopicConstant.HK_ALARM_TOPIC_MESSAGE,containerFactory = "hkListenerContainerFactory")
    public void hkListenPartition5(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                kafkaConsumeService.dealHkMessage(record);
            }
        }
    }

//
//
//    /**
//     * 布尔
//     * @param records
//     */
//    @KafkaListener(id="blCurrent0",topics = KafkaTopicConstant.BL_TOPIC_MESSAGE,containerFactory = "blListenerContainerFactory")
//    public void blListenPartition0(List<ConsumerRecord<?, ?>> records) {
//        for (ConsumerRecord<?, ?> record : records) {
//            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//            if (kafkaMessage.isPresent()) {
//                kafkaConsumeService.dealBLMessage(record);
//            }
//        }
//    }
//
//    /**
//     * 布尔
//     * @param records
//     */
//    @KafkaListener(id="blCurrent1",topics = KafkaTopicConstant.BL_TOPIC_MESSAGE,containerFactory = "blListenerContainerFactory")
//    public void blListenPartition1(List<ConsumerRecord<?, ?>> records) {
//        for (ConsumerRecord<?, ?> record : records) {
//            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//            if (kafkaMessage.isPresent()) {
//                kafkaConsumeService.dealBLMessage(record);
//            }
//        }
//    }
//
//    /**
//     * 布尔
//     * @param records
//     */
//    @KafkaListener(id="blCurrent2",topics = KafkaTopicConstant.BL_TOPIC_MESSAGE,containerFactory = "blListenerContainerFactory")
//    public void blListenPartition2(List<ConsumerRecord<?, ?>> records) {
//        for (ConsumerRecord<?, ?> record : records) {
//            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//            if (kafkaMessage.isPresent()) {
//                kafkaConsumeService.dealBLMessage(record);
//            }
//        }
//    }
//
//
//    /**
//     * 布尔
//     * @param records
//     */
//    @KafkaListener(id="blCurrent3",topics = KafkaTopicConstant.BL_TOPIC_MESSAGE,containerFactory = "blListenerContainerFactory")
//    public void blListenPartition3(List<ConsumerRecord<?, ?>> records) {
//        for (ConsumerRecord<?, ?> record : records) {
//            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//            if (kafkaMessage.isPresent()) {
//                kafkaConsumeService.dealBLMessage(record);
//            }
//        }
//    }
//
//
//    /**
//     * 布尔
//     * @param records
//     */
//    @KafkaListener(id="blCurrent4",topics = KafkaTopicConstant.BL_TOPIC_MESSAGE,containerFactory = "blListenerContainerFactory")
//    public void blListenPartition4(List<ConsumerRecord<?, ?>> records) {
//        for (ConsumerRecord<?, ?> record : records) {
//            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//            if (kafkaMessage.isPresent()) {
//                kafkaConsumeService.dealBLMessage(record);
//            }
//        }
//    }
//
//
//    /**
//     * 布尔
//     * @param records
//     */
//    @KafkaListener(id="blCurrent5",topics = KafkaTopicConstant.BL_TOPIC_MESSAGE,containerFactory = "blListenerContainerFactory")
//    public void blListenPartition5(List<ConsumerRecord<?, ?>> records) {
//        for (ConsumerRecord<?, ?> record : records) {
//            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//            if (kafkaMessage.isPresent()) {
//                kafkaConsumeService.dealBLMessage(record);
//            }
//        }
//    }
//
//
//    /**
//     * 布尔
//     * @param records
//     */
//    @KafkaListener(id="blCurrent6",topics = KafkaTopicConstant.BL_TOPIC_MESSAGE,containerFactory = "blListenerContainerFactory")
//    public void blListenPartition6(List<ConsumerRecord<?, ?>> records) {
//        for (ConsumerRecord<?, ?> record : records) {
//            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//            if (kafkaMessage.isPresent()) {
//                kafkaConsumeService.dealBLMessage(record);
//            }
//        }
//    }
//
//
//    /**
//     * 布尔
//     * @param records
//     */
//    @KafkaListener(id="blCurrent7",topics = KafkaTopicConstant.BL_TOPIC_MESSAGE,containerFactory = "blListenerContainerFactory")
//    public void blListenPartition7(List<ConsumerRecord<?, ?>> records) {
//        for (ConsumerRecord<?, ?> record : records) {
//            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//            if (kafkaMessage.isPresent()) {
//                kafkaConsumeService.dealBLMessage(record);
//            }
//        }
//    }
//
//
//    /**
//     * 布尔
//     * @param records
//     */
//    @KafkaListener(id="blCurrent8",topics = KafkaTopicConstant.BL_TOPIC_MESSAGE,containerFactory = "blListenerContainerFactory")
//    public void blListenPartition8(List<ConsumerRecord<?, ?>> records) {
//        for (ConsumerRecord<?, ?> record : records) {
//            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//            if (kafkaMessage.isPresent()) {
//                kafkaConsumeService.dealBLMessage(record);
//            }
//        }
//    }
//
//
//    /**
//     * 布尔
//     * @param records
//     */
//    @KafkaListener(id="blCurrent9",topics = KafkaTopicConstant.BL_TOPIC_MESSAGE,containerFactory = "blListenerContainerFactory")
//    public void blListenPartition9(List<ConsumerRecord<?, ?>> records) {
//        for (ConsumerRecord<?, ?> record : records) {
//            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//            if (kafkaMessage.isPresent()) {
//                kafkaConsumeService.dealBLMessage(record);
//            }
//        }
//    }
//
//
//    /**
//     * 布尔
//     * @param records
//     */
//    @KafkaListener(id="blCurrent10",topics = KafkaTopicConstant.BL_TOPIC_MESSAGE,containerFactory = "blListenerContainerFactory")
//    public void blListenPartition10(List<ConsumerRecord<?, ?>> records) {
//        for (ConsumerRecord<?, ?> record : records) {
//            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//            if (kafkaMessage.isPresent()) {
//                kafkaConsumeService.dealBLMessage(record);
//            }
//        }
//    }
//
//
//    /**
//     * 布尔
//     * @param records
//     */
//    @KafkaListener(id="blCurrent11",topics = KafkaTopicConstant.BL_TOPIC_MESSAGE,containerFactory = "blListenerContainerFactory")
//    public void blListenPartition11(List<ConsumerRecord<?, ?>> records) {
//        for (ConsumerRecord<?, ?> record : records) {
//            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
//            if (kafkaMessage.isPresent()) {
//                kafkaConsumeService.dealBLMessage(record);
//            }
//        }
//    }

    /**
     * 入侵
     * @param records
     */
    @KafkaListener(id="rqCurrent",topics = KafkaTopicConstant.RQ_ALARM_TOPIC_MESSAGE,containerFactory = "rqListenerContainerFactory")
    public void rqListenPartition(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                kafkaConsumeService.dealRqMessage(record);
            }
        }
    }

    /**
     * 电话
     * @param records
     */
    @KafkaListener(id="dhCurrent",topics = KafkaTopicConstant.DH_ALARM_TOPIC_MESSAGE,containerFactory = "dhListenerContainerFactory")
    public void dhListenPartition(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {

                kafkaConsumeService.dealDhMessage(record);
            }
        }
    }

    /**
     * 门禁
     * @param records
     */
    @KafkaListener(id="mjCurrent",topics = KafkaTopicConstant.MJ_ALARM_TOPIC_MESSAGE,containerFactory = "dhListenerContainerFactory")
    public void mjListenPartition(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                kafkaConsumeService.dealMjMessage(record);
            }
        }
    }



    /**
     * 消防
     * @param records
     */
    @KafkaListener(id="xfCurrent",topics = KafkaTopicConstant.XF_ALARM_TOPIC_MESSAGE,containerFactory = "xfListenerContainerFactory")
    public void xfListenPartition(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                kafkaConsumeService.dealXfMessage(record);
            }
        }
    }

    /**
     * AP
     * @param records
     */
    @KafkaListener(id="apCurrent",topics = KafkaTopicConstant.AP_ALARM_TOPIC_MESSAGE,containerFactory = "apListenerContainerFactory")
    public void apListenPartition(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                kafkaConsumeService.dealApMessage(record);
            }
        }
    }

    /**
     * 采集口
     * @param records
     */
    @KafkaListener(id="gatherCurrent",topics = KafkaTopicConstant.GATHER_ALARM_TOPIC_MESSAGE,containerFactory = "gatherListenerContainerFactory")
    public void gatherListenPartition(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                kafkaConsumeService.dealGatherMessage(record);
            }
        }
    }
}
